package com.xxk.rocket.star.core.consume;

import com.alibaba.rocketmq.client.consumer.DefaultMQPushConsumer;
import com.alibaba.rocketmq.client.consumer.MQConsumer;
import com.alibaba.rocketmq.client.consumer.MQPushConsumer;
import com.alibaba.rocketmq.client.exception.MQClientException;
import com.xxk.rocket.star.core.context.ConsumeContext;
import com.xxk.rocket.star.core.exception.StargateException;
import com.xxk.rocket.star.core.exception.StargateGroupExistException;
import com.xxk.rocket.star.core.utils.ProcessID;
import com.xxk.rocket.star.core.utils.StargateEnvironment;
import com.xxk.rocket.star.core.utils.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

public class ConsumeFactory {

  private static final Map<String, MQPushConsumer> concurrentMap = new ConcurrentHashMap<String, MQPushConsumer>();
  private static final Map<MQPushConsumer, MessageListener> messageListenerMap = new ConcurrentHashMap<MQPushConsumer, MessageListener>();
  private static final Object lock = new Object();
  private static final Logger log = LoggerFactory.getLogger(ConsumeFactory.class);

  public static MQConsumer getConsume(ConsumeContext consumeContext) {
    synchronized (lock) {

      String nameServer = consumeContext.getNamesrvAddr();
      String group = consumeContext.getGroup();
      group = group + "-" + nameServer;

      if (StringUtils.isBlank(nameServer) || StringUtils.isBlank(nameServer) ||
              StargateEnvironment.getNamesrv(nameServer) == null) {
        throw new StargateException("nameServer or group is null");
      }

      MQPushConsumer pushConsumer = concurrentMap.get(group);
      if (pushConsumer != null) {
        //Consumer只能有一个处理方法所以禁止创建两次
        throw new StargateGroupExistException();
      }

      DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(group);
      consumer.setNamesrvAddr(StargateEnvironment.getNamesrv(nameServer));
      consumer.setInstanceName(ProcessID.getPID() + "-" + nameServer);

      concurrentMap.put(group, consumer);
      return consumer;
    }
  }

  public static boolean startMQConsumer(String key) {
    log.info("startMQConsumer:" + key);
    MQPushConsumer mqPushConsumer = concurrentMap.get(key);
    if (mqPushConsumer == null) {
      return false;
    }
    try {
      mqPushConsumer.start();
    } catch (MQClientException e) {
      e.printStackTrace();
      log.error("MQClientException by startMQConsumer:{}", key, e);
      return false;
    }
    log.info("startMQConsumer success:{}", key);
    return true;
  }

  public static void registerMessageListener(MQConsumer mqConsumer,
                                             MessageListener messageListener) {
    MQPushConsumer mqPushConsumer = (MQPushConsumer) mqConsumer;
    mqPushConsumer.registerMessageListener(messageListener);
    messageListenerMap.put(mqPushConsumer, messageListener);
  }

  public static Map<String, MQPushConsumer> getConcurrentMap() {
    return concurrentMap;
  }

  public static MQPushConsumer getConcurrentByGroupAndNameServer(String key) {
    return concurrentMap.get(key);
  }

  public static MessageListener getMessageListenerByConsumer(MQPushConsumer mqPushConsumer) {
    return messageListenerMap.get(mqPushConsumer);
  }
}
